package com.zghhost.codedemo.pulsar;

import org.apache.pulsar.client.api.*;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;

/**
 * @author guohua.zhang@zhaopin.com.cn
 **/
public class ConsumerTest {
    static String url = "pulsar://172.17.5.175:6650";
    static String topic = "persistent://middle/campus-position-merge/job.pulsartest";
    static String subscriptionName = "subscriptionName";
    static SubscriptionType subscriptionType = SubscriptionType.Shared;
    static int timeoutSecond = 1;

    public static void main(String[] args)throws Exception{
        PulsarClient client = client();
        Consumer<byte[]> c1 = consumer(client);
        Consumer<byte[]> c2 = consumer(client);

        BiFunction func = new BiFunction<Message<byte[]>, Throwable, Object>() {
            @Override
            public Object apply(Message<byte[]> message, Throwable throwable) {
                try{
                    System.out.println("c1>"+new String(message.getValue()));
                    Thread.sleep(200);
                }catch (Exception e){
                    e.printStackTrace();
                }
                return null;
            }
        };

        while (true){
            CompletableFuture<Message<byte[]>> m1 = c1.receiveAsync();
            m1.handleAsync(func);
        }

//        CompletableFuture<Message<byte[]>> m2 = c2.receiveAsync();
//        m2.handleAsync(func);

//        System.out.println("程序启动完毕");
    }

    static void processAsync(Consumer<byte[]> c1,BiFunction biFunction){
        CompletableFuture<Message<byte[]>> cf = c1.receiveAsync();
    }

    private static PulsarClient client()throws Exception{
        ClientBuilder clientBuilder = PulsarClient.builder();
        clientBuilder = clientBuilder.serviceUrl(url);
        PulsarClient client = clientBuilder
                .maxNumberOfRejectedRequestPerConnection(50)
                .build();
        return client;
    }

    private static Consumer<byte[]> consumer(PulsarClient client)throws Exception{
        Consumer<byte[]> consumer = null;
        ConsumerBuilder<byte[]> builder = client.newConsumer().topic(topic)
                .subscriptionName(subscriptionName)
                .subscriptionType(subscriptionType)
                .ackTimeout(timeoutSecond, TimeUnit.SECONDS);

            builder.deadLetterPolicy(DeadLetterPolicy.builder()
                    .maxRedeliverCount(2)
                    .deadLetterTopic(topic)
                    .build());

        return builder.subscribe();
    }
}
